package com.icarbonx.dap.notification.util

import java.sql.{DriverManager, PreparedStatement, ResultSet}

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

import scala.collection.mutable

/**
 * Created by Lenovo on 2018/8/28.
 */
class MysqlReceiver(val jdbcURL: String, val name: String, val passwd: String) extends
Receiver[(String, (String, Long))](StorageLevel.MEMORY_AND_DISK_2) {
  def onStart() {
    // Start the thread that receives data over a connection
    new Thread("Socket Receiver") {
      override def run() {
        receive()
      }
    }.start()
  }

  def onStop() {
    // There is nothing much to do as the thread calling receive()
    // is designed to stop by itself isStopped() returns false
  }

  /** Create a socket connection and receive data until receiver is stopped */
  private def receive() {

    try {
      //      val connection = DriverManager.getConnection("jdbc:mysql://10.100.0.9:3306/dap?" +
      //        "useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true", "root", "abcd123")
      val connection = DriverManager.getConnection(jdbcURL, name, passwd)

      val yesterday = DateUtil.getYesterdayFormat()
      val preparedStatement: PreparedStatement = connection.prepareStatement(
        "select * from medication_template_record where actual_time >= '" + yesterday + "'")
      val result: ResultSet = preparedStatement.executeQuery()

      while (result.next()) {
        val person_id = result.getLong("person_id")
        val detail_id = result.getLong("detail_id")
        val actual_time = result.getTimestamp("actual_time")
        val timeFormat = DateUtil.getTimeStampFormat(actual_time.getTime, "yyyy-MM-dd HH:mm:ss")
//        println(s"___ pid $person_id  timeFormat  $timeFormat  ${actual_time.getTime} ")
        store((person_id.toString, (timeFormat, actual_time.getTime)))
      }
      restart("Trying to connect again")
    } catch {
      case e: java.net.ConnectException =>
        restart("Error connecting to ", e)
      case t: Throwable =>
        restart("Error receiving data", t)
    }
  }
}
